package com.ds.infrastructure.hyperspace.container.endpoint.destination;

import com.ds.infrastructure.hyperspace.common.config.Http2Config;
import com.ds.infrastructure.hyperspace.common.event.DataFlushEvent;
import com.ds.infrastructure.hyperspace.container.endpoint.HyperspaceSuspendDataWrapper;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.NettyEndpointNodeStream;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.http.NettyHttpHeaderOptimizer;
import com.ds.infrastructure.hyperspace.container.internal.http2.*;
import com.ds.infrastructure.hyperspace.container.internal.util.Http2Util;
import com.ds.infrastructure.hyperspace.container.internal.util.NettyReleaseUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.DefaultHttp2SettingsFrame;
import io.netty.handler.codec.http2.Http2HeadersFrame;
import io.netty.handler.codec.http2.Http2Settings;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import static com.ds.infrastructure.hyperspace.common.StateEnum.*;
import static com.ds.infrastructure.hyperspace.common.TargetEndpointStreamStatus.CLOSED;
import static com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameTypeEnum.*;

/**
 * @author  Weishaoying
 */
@Slf4j
public class Http2TargetProtocolAdapter extends AbstractTargetProtocolAdapter {

    private final ArrayDeque<Http2Settings> outstandingLocalSettingsQueue =
            new ArrayDeque<>(4);

    private final HyperspaceHttp2Codec hyperspaceHttp2Codec;

    private final Http2Config http2Config;

    private HyperspaceDuplexFlowContext connectionFlowContext;

    private HyperspaceDuplexFlowContext currentFlowContext;

    private boolean connectionReady;

    private List<HyperspaceSuspendDataWrapper> suspendRequestList = new ArrayList<>();

    private List<HyperspaceHttp2FrameWrapper> suspendResponseList = new ArrayList<>();

    private int dataFramePayloadSize = 0;

    private boolean isAutoFlushData = true;

    public Http2TargetProtocolAdapter(NettyContainer nettyContainer,
                                      HyperspaceConnector hyperspaceConnector,
                                      TargetEndpointNode targetEndpointNode,
                                      HyperspaceHttp2Codec hyperspaceHttp2Codec) {
        super(nettyContainer, hyperspaceConnector, targetEndpointNode);
        this.hyperspaceHttp2Codec = hyperspaceHttp2Codec;
        this.http2Config = nettyContainer.getHttp2Config();
    }

    @Override
    public void handleChannelRead(ChannelHandlerContext ctx, Object msg) {
        stream.touch();
        HyperspaceHttp2FrameWrapper http2FrameWrapper = (HyperspaceHttp2FrameWrapper) msg;
        HyperspaceHttp2FrameTypeEnum frameType = http2FrameWrapper.getFrameType();
        if (http2FrameWrapper.getStreamId() == 0 || CLOSED == stream.getStatus()) {
            if (WINDOW_UPDATE == frameType) {
                return;
            }
            if (GO_AWAY == frameType) {
                HyperspaceHttp2GoAwayFrame hyperspaceHttp2GoAwayFrame =
                        (HyperspaceHttp2GoAwayFrame) http2FrameWrapper.getFrameData();
                hyperspaceHttp2GoAwayFrame.getOriginGoAwayFrame().release();
                closeTargetChannel();
                return;
            }
            if (CLOSED == stream.getStatus()) {
                releaseData(http2FrameWrapper);
                return;
            }
            if (SETTINGS == frameType) {
                if (connectionReady) {
                    writeDataToEntry(http2FrameWrapper);
                } else {
                    outstandingLocalSettingsQueue.add(((DefaultHttp2SettingsFrame) http2FrameWrapper.getFrameData()).settings());
                }
                return;
            }
            if (!connectionReady && SETTINGS_ACK == frameType) {
                applySettings(ctx, http2FrameWrapper);
                connectionReady = true;
                hyperspaceHttp2Codec.writeAndFlushWindowUpdateFrame(ctx, http2Config.getWindowUpdateInitIncrement());
                return;
            }
            if (PING == frameType || PING_ACK == frameType) {
                if (connectionFlowContext != null) {
                    targetFlowOperation.handleOutput(connectionFlowContext, http2FrameWrapper);
                }
                return;
            }
        }
        writeDataToEntry(http2FrameWrapper);
        if (currentFlowContext == null) {
            return;
        }
        if (RST_STREAM == frameType) {
            targetFlowOperation.interruptAndCloseContext(currentFlowContext, TARGET_STREAM_CANCEL
                    , "Target stream cancel,received Rst Stream");
            return;
        }
        //END_HEADERS
        if (http2FrameWrapper.isEndOfStream()) {
            if (HEADERS == frameType || Http2Util.isErrorCode(currentFlowContext.getTargetResponseCode())) {
                currentFlowContext.touch();
                currentFlowContext.setOutputComplete(true);
                targetFlowOperation.handleOutputComplete(currentFlowContext);
            }
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        if (currentFlowContext != null) {
            if (currentFlowContext.isOutputComplete() || currentFlowContext.isClosed()) {
                return;
            }
            currentFlowContext.touch();
            targetFlowOperation.handleOutputComplete(currentFlowContext);
        }
    }

    @Override
    public void resetChannelActiveState(ChannelHandlerContext channelHandlerContext,
                                        NettyEndpointNodeStream stream,
                                        HyperspaceDuplexFlowContext duplexFlowContext) {
        super.resetChannelActiveState(channelHandlerContext, stream, duplexFlowContext);
        this.connectionReady = true;
        this.connectionFlowContext = duplexFlowContext;
        hyperspaceHttp2Codec.sendPreface(channelHandlerContext,
                targetEndpointNode.getNodeConfig().getProtocolConfig());
    }

    @Override
    public void writeDataToTarget(HyperspaceDuplexFlowContext duplexFlowContext, Object msg) {
        if (isChannelClosedReleaseData(duplexFlowContext, msg)) {
            return;
        }
        NettyHttpHeaderOptimizer.optimizeHttp2Header(duplexFlowContext, msg, targetEndpointNode);
        duplexFlowContext.setTargetRemoteAddress(targetRemoteAddress);
        if (duplexFlowContext.getRequestStartState() == msg && duplexFlowContext.getTracerContext() != null) {
            HyperspaceHttp2FrameWrapper http2FrameWrapper = (HyperspaceHttp2FrameWrapper) msg;
            Http2HeadersFrame http2HeadersFrame = (Http2HeadersFrame) http2FrameWrapper.getFrameData();
            duplexFlowContext.getTracerContext().inject(targetRemoteAddress, http2HeadersFrame.headers());
        }
        if (connectionReady) {
            sendSuspendRequestData();
            writeHttp2Data(channelHandlerContext, msg);
        } else {
            suspendRequestList.add(new HyperspaceSuspendDataWrapper(duplexFlowContext, msg, false));
        }
    }

    private void writeHttp2Data(ChannelHandlerContext channelHandlerContext, Object msg) {
        HyperspaceHttp2FrameWrapper http2FrameWrapper = (HyperspaceHttp2FrameWrapper) msg;
        if (DATA == http2FrameWrapper.getFrameType()) {
            dataFramePayloadSize += http2FrameWrapper.payloadSize();
            if (dataFramePayloadSize > http2Config.getWindowUpdateSendIncrement()) {
                hyperspaceHttp2Codec.writeAndFlushWindowUpdateFrame(channelHandlerContext, dataFramePayloadSize);
                dataFramePayloadSize = 0;
            }
        }
        hyperspaceHttp2Codec.writeData(channelHandlerContext, msg);
        if (isAutoFlushData) {
            isAutoFlushData = false;
            channel.eventLoop().execute(() -> {
                userEventTriggered(channelHandlerContext, DataFlushEvent.FLUSH_EVENT);
            });
        }
    }

    @Override
    public HyperspaceDuplexFlowContext connectionDuplexFlowContext() {
        return connectionFlowContext;
    }

    @Override
    public boolean requireDuplexFlowContext(HyperspaceDuplexFlowContext duplexFlowContext) {
        try {
            if (connectionFlowContext == null) {
                connectionFlowContext = duplexFlowContext;
                List<Object> prefaceDataList = (List<Object>) duplexFlowContext.getEntryPrefaceState();
                if (!CollectionUtils.isEmpty(prefaceDataList)) {
                    for (Object prefaceData : prefaceDataList) {
                        if (!isChannelActive()) {
                            log.warn("Init channel state but channel close Unexpected! host {}, url {}",
                                    duplexFlowContext.getHost(), duplexFlowContext.getUri());
                            break;
                        }
                        hyperspaceHttp2Codec.writeData(channelHandlerContext, prefaceData);
                    }
                    channelHandlerContext.flush();
                }
            }
            return true;
        } catch (Exception e) {
            targetFlowOperation.interruptAndCloseContext(duplexFlowContext, GATEWAY_ERROR,
                    "DuplexFlowContext state incorrect, target node " +
                            targetEndpointNode.getKey() + ",Exception:" + e);
            throw e;
        }
    }

    private void applySettings(ChannelHandlerContext ctx, HyperspaceHttp2FrameWrapper http2SettingAckFrameWrapper) {
        Http2Settings http2Settings = outstandingLocalSettingsQueue.poll();
        if (http2Settings != null) {
            hyperspaceHttp2Codec.writeAndFlushSettingsAck(ctx);
            hyperspaceHttp2Codec.applySettings(http2Settings);
            if (connectionFlowContext.getConnectionDuplexFlowContext() == null) {
                HyperspaceHttp2SettingAckFrame settingAckFrame =
                        (HyperspaceHttp2SettingAckFrame) http2SettingAckFrameWrapper.getFrameData();
                settingAckFrame.setSettings(http2Settings);
                targetFlowOperation.handleOutput(connectionFlowContext, http2SettingAckFrameWrapper);
            }
        }
        sendSuspendRequestData();
    }

    private void sendSuspendRequestData() {
        if (connectionFlowContext == null || suspendRequestList.isEmpty()) {
            return;
        }
        Iterator<HyperspaceSuspendDataWrapper> iterator = suspendRequestList.iterator();
        while (iterator.hasNext()) {
            try {
                HyperspaceSuspendDataWrapper suspendSendWrapper = iterator.next();
                HyperspaceDuplexFlowContext duplexFlowContext = suspendSendWrapper.getDuplexFlowContext();
                if (duplexFlowContext.isClosed() &&
                        suspendSendWrapper.getData() == duplexFlowContext.getRequestStartState()) {
                    continue;
                }
                if (duplexFlowContext.isClosed() || !duplexFlowContext.getEntryProtocolAdapter().isChannelOpen()) {
                    releaseData(suspendSendWrapper.getData());
                    continue;
                }
                if (isChannelClosedReleaseData(duplexFlowContext, suspendSendWrapper.getData())) {
                    log.warn("Try to send suspend request data, but channel is close!");
                    return;
                }
                writeHttp2Data(channelHandlerContext, suspendSendWrapper.getData());
            } finally {
                iterator.remove();
            }
        }
        channelHandlerContext.flush();
    }

    private void writeDataToEntry(HyperspaceHttp2FrameWrapper http2FrameWrapper) {
        currentFlowContext = findDuplexFlowContext(http2FrameWrapper);
        if (currentFlowContext == null) {
            if (http2FrameWrapper.getStreamId() == 0) {
                suspendResponseList.add(http2FrameWrapper);
            } else {
                releaseData(http2FrameWrapper);
            }
            return;
        }
        if (!CollectionUtils.isEmpty(suspendResponseList)) {
            Iterator<HyperspaceHttp2FrameWrapper> iterator = suspendResponseList.iterator();
            while (iterator.hasNext()) {
                HyperspaceHttp2FrameWrapper suspendReadHttp2FrameWrapper = iterator.next();
                targetFlowOperation.handleOutput(
                        findDuplexFlowContext(suspendReadHttp2FrameWrapper),
                        suspendReadHttp2FrameWrapper);
                iterator.remove();
            }
        }
        targetFlowOperation.handleOutput(currentFlowContext, http2FrameWrapper);
    }

    private HyperspaceDuplexFlowContext findDuplexFlowContext(HyperspaceHttp2FrameWrapper frameWrapper) {
        if (connectionFlowContext == null) {
            return null;
        }
        if (frameWrapper.getStreamId() > 1) {
            return connectionFlowContext.getEntryProtocolAdapter()
                    .requireDuplexFlowContext(frameWrapper.getStreamId());
        }
        return connectionFlowContext;
    }

    @Override
    public void releaseData(Object data) {
        if (data instanceof HyperspaceHttp2FrameWrapper) {
            HyperspaceHttp2FrameWrapper hyperspaceHttp2FrameWrapper = (HyperspaceHttp2FrameWrapper) data;
            NettyReleaseUtil.releaseData(hyperspaceHttp2FrameWrapper.getFrameData());
        }
    }

    private boolean isChannelClosedReleaseData(HyperspaceDuplexFlowContext duplexFlowContext, Object msg) {
        if (!isChannelActive()) {
            if (msg == duplexFlowContext.getRequestStartState()) {
                targetFlowOperation.interruptAndCloseContext(duplexFlowContext, CLIENT_CLOSED_CONNECTION,
                        "Target client has closed connection,remote address:" + targetRemoteAddress);
            } else {
                releaseData(msg);
            }
            return true;
        }
        return false;
    }

    @Override
    public void clearSuspendData() {
        Iterator<HyperspaceSuspendDataWrapper> suspendRequestDataWrapperIterator = suspendRequestList.iterator();
        while (suspendRequestDataWrapperIterator.hasNext()) {
            HyperspaceSuspendDataWrapper suspendRequestDataWrapper = suspendRequestDataWrapperIterator.next();
            if (suspendRequestDataWrapper.getDuplexFlowContext().getRequestStartState()
                    != suspendRequestDataWrapper.getData()) {
                releaseData(suspendRequestDataWrapper.getData());
            }
            suspendRequestDataWrapperIterator.remove();
        }
        Iterator<HyperspaceHttp2FrameWrapper> suspendResponseDataWrapperIterator = suspendResponseList.iterator();
        while (suspendResponseDataWrapperIterator.hasNext()) {
            HyperspaceHttp2FrameWrapper suspendResponseDataWrapper = suspendResponseDataWrapperIterator.next();
            releaseData(suspendResponseDataWrapper);
            suspendResponseDataWrapperIterator.remove();
        }
        if (stream != null) {
            stream.clearSuspendRequest();
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof DataFlushEvent) {
            if (ctx.channel().isActive()) {
                ctx.flush();
                isAutoFlushData = true;
            }
            return;
        }
        ctx.fireUserEventTriggered(evt);
    }

}
